package worldCount;


import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

import java.io.IOException;

public class MyMapReduceDriver {
    public static void main(String[] args) throws IOException, InterruptedException, ClassNotFoundException {
        Path Inputpath = new Path("D:\\MyProject\\gmall\\example\\world");
        Path Outputpath = new Path("D:\\MyProject\\gmall\\example\\result");

        // 做job任务的配置
        Configuration conf = new Configuration();

//        检查路径是否存在
        FileSystem fs = FileSystem.get(conf);
        if(fs.exists(Outputpath)){
            fs.delete(Outputpath,true);
        }

        //  创建job任务
        Job job = Job.getInstance(conf);
        // 将某个类所在地jar包作为job的jar包
        job.setJarByClass(MyMapReduceDriver.class);

        // 为Job创建一个名字
        job.setJobName("wordCount");

        // ②设置Job
        // 设置Job运行的Mapper，Reducer类型，Mapper,Reducer输出的key-value类型
        job.setMapperClass(MyMapper.class);
        job.setReducerClass(MyReduce.class);
        // Job需要根据Mapper和Reducer输出的Key-value类型准备序列化器，通过序列化器对输出的key-value进行序列化和反序列化
        // 如果Mapper和Reducer输出的Key-value类型一致，直接设置Job最终的输出类型
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(IntWritable.class);

        //  设置输入目录和输出目录
        FileInputFormat.setInputPaths(job,Inputpath);
        FileOutputFormat.setOutputPath(job,Outputpath);
        //③运行Job
        job.waitForCompletion(true);





    }
}
